package task1

import org.apache.spark.internal.Logging
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable

/**
 * 倒排索引作业
 */
object InvertedIndex extends Logging {

  def main(args: Array[String]): Unit = {
    // 构建 Spark 环境
    val conf = new SparkConf()
      .setIfMissing("spark.master", "local[2]")
      .setAppName("InvertedIndex")
    val sc = new SparkContext(conf)

    val filePath = getClass.getResource("/").getPath + "*txt"
    logInfo(s"读取资源文件：$filePath")
    // 使用 wholeTextFiles 读取文件时将文件名作为 key 值
    val invertedIndex = sc.wholeTextFiles(filePath)
      .flatMap { case (filename, fileText) =>
        // 切割文本内容并将单词与截取的文件名组成键值对
        val start = filename.lastIndexOf("/") + 1
        fileText.split(" ").map((_, filename.substring(start, start + 1)))
      }.aggregateByKey(mutable.HashMap[String, Int]())(
      // 对于单个 partition，使用 Map 记录词频
      (cntMap, filename) => {
        cntMap.update(filename, cntMap.getOrElse(filename, 0) + 1)
        cntMap
      },
      // 将 partition 之间的结果合并
      (cnt1, cnt2) => {
        cnt1.foreach(kv => cnt2.update(kv._1, cnt1(kv._1) + cnt2.getOrElse(kv._1, 0)))
        cnt2
      }).collect()

    // 打印结果
    invertedIndex.sortBy(_._1).foreach(x => println(x._1, x._2.toSeq.sorted))

  }
}
